package org.budo.dubbo.registry.redis.util;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.budo.dubbo.registry.redis.AbstractBudoRedisRegistry;
import org.budo.dubbo.registry.redis.BudoRedisRegistry;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.UrlUtils;

import redis.clients.jedis.Jedis;

/**
 * @author lmw
 */
public class BudoRedisRegistryRepository {
    private static final Logger log = LoggerFactory.getLogger(BudoRedisRegistryRepository.class);

    /**
     * 根据接口查询Provider
     */
    public List<URL> lookup(BudoRedisRegistry budoRedisRegistry, URL url) {
        String serviceInterface = url.getServiceInterface();

        String key = budoRedisRegistry.getProtocol() + ":" + serviceInterface + Constants.PATH_SEPARATOR + Constants.PROVIDERS_CATEGORY;
        Map<String, String> map = this.hGetAllByKey(budoRedisRegistry, key); // 去Redis中查询

        return this.nonRepeatingLookupUrls(map);
    }

    /**
     * 查询匹配的地址
     */
    public List<URL> listMatchUrlsForProcessMessage(Jedis jedis, String notifyKey, URL notifyListenerUrl) {
        long now = System.currentTimeMillis();
        ArrayList<URL> notifyUrls = new ArrayList<URL>();

        // hgetAll
        Map<String, String> hgetAllValues = jedis.hgetAll(notifyKey);

        if (hgetAllValues != null && hgetAllValues.size() > 0) {
            for (Map.Entry<String, String> hgetEntry : hgetAllValues.entrySet()) {
                URL hgetKeyUrl = URL.valueOf(hgetEntry.getKey());
                long expire = Long.parseLong(hgetEntry.getValue());
                if (!hgetKeyUrl.getParameter(Constants.DYNAMIC_KEY, true) //
                        || expire >= now) { // 时间未过期
                    if (UrlUtils.isMatch(notifyListenerUrl, hgetKeyUrl)) {
                        notifyUrls.add(hgetKeyUrl);
                    }
                }
            }
        }
        return notifyUrls;
    }

    public void unregister(BudoRedisRegistry budoRedisRegistry, URL url) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);

            String key = DubboUrlUtil.toCategoryPath(budoRedisRegistry, url);
            String value = url.toFullString();
            jedis.hdel(key, value);
        } catch (Throwable e) {
            log.error("unregister error, error=" + e, e);
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    public void publishUnregister(BudoRedisRegistry budoRedisRegistry, URL url) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);

            String key = DubboUrlUtil.toCategoryPath(budoRedisRegistry, url);
            jedis.publish(AbstractBudoRedisRegistry.PUBLISH_CHANNEL_PREFIX + key, Constants.UNREGISTER);
        } catch (Throwable e) {
            log.error("publishUnregister error, error=" + e, e);
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    /**
     * @return 如果字段是哈希表中的一个新建字段，并且值设置成功，返回 1 。 如果哈希表中域字段已经存在且旧值已被新值覆盖，返回 0 。
     */
    public long register(BudoRedisRegistry budoRedisRegistry, URL url) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);

            String key = DubboUrlUtil.toCategoryPath(budoRedisRegistry, url);
            String field = url.toFullString();
            String value = String.valueOf(System.currentTimeMillis() + 60 * 1000); // expire

            return jedis.hset(key, field, value);
        } catch (Throwable e) {
            log.error("#105 register error, error=" + e, e);
            return -1;
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    public void publishRegister(BudoRedisRegistry budoRedisRegistry, URL url) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);

            String key = DubboUrlUtil.toCategoryPath(budoRedisRegistry, url);
            String channel = AbstractBudoRedisRegistry.PUBLISH_CHANNEL_PREFIX + key;

            jedis.publish(channel, Constants.REGISTER);
        } catch (Throwable e) {
            log.error("doPublishRegister error, error=" + e, e);
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    public Boolean isAvailable(BudoRedisRegistry budoRedisRegistry) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);
            return jedis.isConnected();
        } catch (Throwable e) {
            log.error("isAvailable error, this=" + this + ", e=" + e, e);
            return false;
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    public void cleanExpiredKeys(BudoRedisRegistry budoRedisRegistry, URL url) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);

            String key = DubboUrlUtil.toCategoryPath(budoRedisRegistry, url);
            Map<String, String> values = jedis.hgetAll(key);
            if (values != null && values.size() > 0) {
                Boolean delete = this.deleteExpiredKey(jedis, key, values);
                if (delete) {
                    jedis.publish(AbstractBudoRedisRegistry.PUBLISH_CHANNEL_PREFIX + key, Constants.UNREGISTER);
                }
            }
        } catch (Throwable e) {
            log.error("cleanExpiredKeys error, this=" + this + ", e=" + e, e);
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    /**
     * 用Map排除重复
     */
    private List<URL> nonRepeatingLookupUrls(Map<String, String> map) {
        Map<String, URL> notifyUrlMap = new HashMap<String, URL>();
        for (Entry<String, String> entry : map.entrySet()) {
            String entryKey = entry.getKey();
            URL entryUrl = URL.valueOf(entryKey);
            String urlKey = DubboUrlUtil.toSimpleString(entryUrl);
            notifyUrlMap.put(urlKey, entryUrl);
        }

        Set<Entry<String, URL>> entrySet = notifyUrlMap.entrySet();

        // 返回非重复的
        List<URL> notifyUrls = new ArrayList<URL>();
        for (Entry<String, URL> entry : entrySet) {
            URL value = entry.getValue();
            notifyUrls.add(value);
        }

        return notifyUrls;
    }

    private Map<String, String> hGetAllByKey(BudoRedisRegistry budoRedisRegistry, String key) {
        Jedis jedis = null;
        try {
            jedis = RegistryJedisUtil.getJedisResource(budoRedisRegistry);
            Map<String, String> values = jedis.hgetAll(key);
            return values;
        } catch (Throwable e) {
            log.error("hgetAll error, key=" + key + ", e=" + e, e);
            return new HashMap<String, String>();
        } finally {
            RegistryJedisUtil.returnResource(budoRedisRegistry, jedis);
        }
    }

    private Boolean deleteExpiredKey(Jedis jedis, String key, Map<String, String> values) {
        long now = System.currentTimeMillis();

        Boolean delete = false;
        for (Map.Entry<String, String> entry : values.entrySet()) {
            URL url = URL.valueOf(entry.getKey());
            if (url.getParameter(Constants.DYNAMIC_KEY, true)) {
                long expire = Long.parseLong(entry.getValue());
                if (expire < now) {
                    jedis.hdel(key, entry.getKey());
                    delete = true;

                    if (log.isDebugEnabled()) {
                        log.debug("Delete expired key: " + key + " -> value: " + entry.getKey() + ", expire: " + new Date(expire) + ", now: " + new Date(now));
                    }
                }
            }
        }
        return delete;
    }
}